#include <sstream>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <sys/time.h>
#include <sys/stat.h>
#include <glib.h>
#include "MemCache.h"
#include "Debug.h"

#define CACHE_HOST_MAX         256
#define MEMCACHE_LISTEN_PORT   11211
using namespace panda;

bool MemCache:: Initialize(const char *conf)
{
	struct stat st;
	if (lstat(conf, &st) != 0)
	{
		cerr<<__FILE__<<":"<<__func__<<":"<<__LINE__<<" file '"<<conf<<"' is not exist!"<<endl;
		return false;
	}

	GKeyFile *kv = g_key_file_new();
	GError *err = NULL;
	if(TRUE != g_key_file_load_from_file(kv,conf, G_KEY_FILE_NONE , &err))
	{
		if (err != NULL)
		{
			cerr << "Unable to read file:" << err->message << endl;
			g_error_free (err);
		}
		g_key_file_free(kv);
		return false;
	}

	gchar *pval =  g_key_file_get_value(kv,"global", "farms",  NULL);
	if(pval == NULL)
	{
		cerr<<"no keys in global"<< endl;	
		g_key_file_free(kv);
		return false;
	}

	string subs = pval;
	g_free(pval);
	char **columns = g_strsplit(subs.c_str(), ",", -1);
	int columns_num =  g_strv_length(columns);
	if(columns_num > CACHE_HOST_MAX )
	{
		cerr<<"nodes are too much !"<<endl;
                g_strfreev(columns);         
		g_key_file_free(kv);
		return false;
	}

	int *pIA = new int[ CACHE_HOST_MAX];
	memset(pIA, -1, CACHE_HOST_MAX );

	for(int i=0; i< columns_num ; i++ )
	{

		pval =  g_key_file_get_value(kv, g_strstrip(columns[i]), "key",  NULL);
		if(pval == NULL)
		{
			cerr<<"no keys in global"<< endl;	
			g_strfreev(columns);         
			g_key_file_free(kv);
                        delete []pIA;
			return false;
		}
		string sk  = pval;
		g_free(pval); 

		pval =  g_key_file_get_value(kv, g_strstrip(columns[i]), "host",  NULL);
		if(pval == NULL)
		{
			cerr<<"no keys in global"<< endl;	
			g_strfreev(columns);         
			g_key_file_free(kv);
                        delete []pIA;
			return false;
		}
		string hName = g_strstrip(pval);
		g_free(pval); 

		pval =  g_key_file_get_value(kv,g_strstrip(columns[i]), "port",  NULL);
		if(pval == NULL)
		{
			cerr<<"no keys in global"<< endl;
			g_strfreev(columns);         
			g_key_file_free(kv);
                        delete []pIA;
			return false;
		}
		string sp = pval;
                g_free(pval); 

		pval =  g_key_file_get_value(kv,g_strstrip(columns[i]), "backup",  NULL);
		if(pval == NULL)
		{
			cerr<<"no keys in global"<< endl;	
			g_strfreev(columns);         
			g_key_file_free(kv);
                        delete []pIA;
			return false;
		}
		string sb = g_strstrip(pval);
                g_free(pval); 
		int iport =  atoi(sp.c_str());
		int ikey = atoi(sk.c_str());
		int iback = atoi(sb.c_str());

		if(pIA[ikey] != ikey)
		{
			pIA[ikey] = ikey;
			MemConn *pMc = new MemConn;
			pMc->pCon = new MemConnection(hName.c_str(),iport, ikey, iback);
#ifdef _PTHREAD_SAFE
			pthread_mutex_init(&pMc->lock, NULL);
#endif
			m_Conn_Map[ikey] = pMc;
		}
	}
	g_strfreev(columns);         
	g_key_file_free(kv);

	bool bret = true;
	for(int i=0; i< columns_num; i++)
	{
		if(pIA[i] != i)
		{
			cerr<<__FILE__<<":"<<__func__<<":"<<"key = "<<i<<" has not be defined!"<<endl;
			bret = false;
			break;      
		}	
	}
	DEBUG1("key's define is ok");
	if(bret)
		for(int i=0; i< columns_num; i++)
		{
			MemConn *p = (MemConn *) m_Conn_Map[i];
			if(p->pCon==NULL || p->pCon->m_iBackup <0 || p->pCon->m_iBackup >= columns_num)
			{
				cerr<<i<<" 's backup ="<< p->pCon->m_iBackup<<"; but this key is not exist!"<<endl;
				bret = false;
				break;      
			}		 
		}
	if(bret)
	{
		DEBUG1("now, let's connect!");
		for(int i=0; i< columns_num; i++)
		{
			MemConn *p = (MemConn *) m_Conn_Map[i];
			if(p!= NULL && p->pCon != NULL )
			{
				DEBUG2("Yes, Connect to ", i);
				p->pCon->Connect();
			}
		}
	}

	delete []pIA;
	if(!bret)
	{
		DEBUG1("is false, why?");
		map<int, MemConn *> :: iterator iter;
		for(iter=m_Conn_Map.begin(); iter!= m_Conn_Map.end(); iter++)
		{
			MemConn *p = iter->second;
			if(p != NULL )
			{
				DEBUG1("delete pCon");
#ifdef _PTHREAD_SAFE	
				pthread_mutex_destroy(&p->lock);
#endif	
				if(p->pCon != NULL)
					delete p->pCon;
				p->pCon = NULL;
				delete p;
				p = NULL;
			}
		}
		m_Conn_Map.clear();		
	}
	m_iPartition = m_Conn_Map.size();
	DEBUG2("new connect size = ", m_iPartition);
	return bret;
}	

 MemCache :: ~MemCache( )
{	
	map<int, MemConn *> :: iterator iter;
	for(iter=m_Conn_Map.begin(); iter!= m_Conn_Map.end(); iter++)
	{
		MemConn *p = iter->second;
		if(p != NULL )
		{
#ifdef _PTHREAD_SAFE	
			pthread_mutex_destroy(&p->lock);
#endif	
			if(p->pCon != NULL)
				delete p->pCon;
			p->pCon = NULL;
			delete p;
			p = NULL;
		}
	}
	m_Conn_Map.clear();		
}

int MemCache :: Set(const char *data, unsigned dlen) //, char **rval,unsigned &wsize, unsigned &rlen)
{ //  hash + expire + argc + key-val.key-val...
        unsigned uint_len = sizeof(uint32_t);
        if( dlen <= uint_len*5 )
        {
              DEBUG4("dlen <= uint_len*5 is true : dlen =", dlen, "; uint_len*5=", uint_len*5);
              return PANDA_ERR_INPUT_DATA; 
        }
	unsigned ilen = 0;
        int ihash = 0;
	bcopy(data, &ihash, uint_len);
	ilen += uint_len;
        DEBUG4("ihash = ", ihash, "; offset = ", ilen); 
        long expire = 0;
       	bcopy(data + ilen, &expire, sizeof(long));
        ilen += sizeof(long);
        DEBUG4("expire = ", expire, "; offset = ", ilen + 4 - sizeof(long)); 
        int argc = 0;
	bcopy(data + ilen , &argc, uint_len);
	ilen += uint_len;
        DEBUG4("argc = ", argc, "; offset = ", ilen); 
        if( argc <= 0)
        {
              DEBUG2("argc=0, means no key=value to save! ilen =", ilen); 
              return PANDA_ERR_INPUT_DATA; 
        }
        DEBUG6("ihash=",ihash,"; expire=",expire,"; argc=",argc);
	int idx = ihash % m_iPartition;
	int iret = PANDA_MC_FAIL;
	for(int i=0; i< argc; i++)
	{
		int klen ,vlen;
		bcopy(data + ilen, &klen, uint_len);
		ilen += uint_len;
		if(ilen + klen >= dlen || klen > PANDA_MC_KEY_MAX )
                {
                        DEBUG4("ilen+klen>=dlen means no value; ilen=", ilen, "; klen=",klen); 
			return PANDA_ERR_INPUT_DATA;       
                }
		char *key = (char *)data + ilen;
                 
                ilen += klen;
		bcopy(data + ilen, &vlen, uint_len);
		ilen += uint_len;
		if(ilen + vlen > dlen || klen > PANDA_MC_VALUE_MAX )
                {
                    DEBUG4("ilen+vlen >dlen means value is error; ilen=", ilen, "; vlen=", vlen);
			return PANDA_ERR_INPUT_DATA;      
                } 
		char *val = (char *)data + ilen;
		ilen += vlen;

		_Lock(idx); 
		MemConn *p = (MemConn *)m_Conn_Map[idx];
		int istatus = p->pCon->getStatus();
		if(istatus == PANDA_MC_RESTART || istatus == PANDA_MC_LIVE )
			istatus = iret = p->pCon->Set(key, klen, val, vlen, expire);
		_UnLock(idx);
		if(PANDA_MC_NEXT == istatus)
		{
			idx = p->pCon->m_iBackup;
			_Lock(idx);
			long lt = p->pCon->getLastTime();
			p = (MemConn *)m_Conn_Map[idx];
			istatus = p->pCon->getStatus();   
			if( istatus == PANDA_MC_LIVE)
			{
				ostringstream os;
				char *nkey = g_strndup((char *)key, klen);
				os<< lt <<"|"<< nkey;
				g_free(nkey);
				istatus = iret = p->pCon->Set(os.str().c_str(), os.str().length(),val, vlen, expire);
			}
			_UnLock(idx);
		}
		if(istatus == PANDA_MC_NEXT)
		{	DEBUG1("all is down, can't use !");  break;  }
	}
	return iret;
}

int MemCache :: Get(const char *data, unsigned dlen, char **rval, unsigned &wsize, unsigned &rlen)
{ // hash + keycount + keys... 
        unsigned uint_len = sizeof(uint32_t);
        if( dlen <= uint_len*3 )
        {
              DEBUG2("dlen <= uint_len*3 is true : dlen =", dlen);
              return PANDA_ERR_INPUT_DATA; 
        }
	unsigned ilen = 0;
        int ihash = 0;
	bcopy(data, &ihash, uint_len);
	ilen += uint_len;
        DEBUG4("ihash = ", ihash, "; offset = ", ilen); 

        int argc = 0;
	bcopy(data + ilen , &argc, uint_len);
	ilen += uint_len;
        DEBUG4("key counts = ", argc, "; offset = ", ilen); 
        if( argc <= 0)
        {
              DEBUG1("argc=0, means no key to get!"); 
              return PANDA_ERR_INPUT_DATA; 
        }
	int idx = ihash % m_iPartition;
	int iret = PANDA_MC_FAIL;

	
	_Lock(idx);
	MemConn *p = (MemConn *)m_Conn_Map[idx];
	int istatus = p->pCon->getStatus();
	if(PANDA_MC_RESTART == istatus || PANDA_MC_LIVE == istatus )
	{
		if(argc == 1)
			istatus = iret = p->pCon->Get(data+ilen, dlen-ilen,NULL, rval, wsize, rlen);
		else
			istatus = iret = p->pCon->MGet(data+ilen, dlen-ilen, argc,NULL, rval,wsize, rlen);
	} 
	_UnLock(idx);
DEBUG2("iret = istatus = ", iret);
	if(PANDA_MC_NEXT == istatus)
	{
DEBUG1("master is down, so try slave!" );
		idx = p->pCon->m_iBackup;
		_Lock(idx);
		long lt = p->pCon->getLastTime();
		p = (MemConn *)m_Conn_Map[idx];
		istatus = p->pCon->getStatus(); 
		if(PANDA_MC_LIVE  == istatus )
		{
			ostringstream os;
			os<< lt<<"|";
                        if(argc == 1)
			istatus = iret = p->pCon->Get(data+ilen, dlen-ilen, os.str().c_str(), rval, wsize, rlen);
                        else
			istatus = iret = p->pCon->MGet(data+ilen, dlen-ilen, argc, os.str().c_str(), rval, wsize, rlen);
		}
		_UnLock(idx);
	}
	if(PANDA_MC_NEXT == istatus)
	{	DEBUG1("all is down, can't use !"); }

	return iret;
}


int MemCache :: Incr(const char *data, unsigned dlen, char **buf,unsigned &wsize, unsigned int &retv)
{ // hash key value
        unsigned uint_len = sizeof(uint32_t);
        if( dlen <= uint_len*2 )
        {
              DEBUG2("dlen <= uint_len*2 is true : dlen =", dlen);
              return PANDA_ERR_INPUT_DATA; 
        }
	unsigned ilen = 0;
        int ihash = 0;
	bcopy(data, &ihash, uint_len);
	ilen += uint_len;
        DEBUG4("ihash = ", ihash, "; offset = ", ilen); 
        
	int idx = ihash % m_iPartition;
	int iret = PANDA_MC_FAIL;
	_Lock(idx); 
	MemConn *p = (MemConn *)m_Conn_Map[idx];
	int istatus = p->pCon->getStatus();
	if(istatus == PANDA_MC_RESTART || istatus == PANDA_MC_LIVE )
	       istatus = iret = p->pCon->Incr(data + ilen, dlen - ilen, NULL , buf, wsize, retv);
	_UnLock(idx);
	if(PANDA_MC_NEXT == istatus)
	{
		idx = p->pCon->m_iBackup;
		_Lock(idx);
		long lt = p->pCon->getLastTime();
		p = (MemConn *)m_Conn_Map[idx];
		istatus = p->pCon->getStatus();   
		if( istatus == PANDA_MC_LIVE)
		{
			ostringstream os;
			os<< lt <<"|";
			istatus=iret=p->pCon->Incr(data+ilen, dlen-ilen, os.str().c_str(), buf, wsize, retv);
		}
		_UnLock(idx);
	}
	if(istatus == PANDA_MC_NEXT)
	{	DEBUG1("all is down, can't use !"); }
	return iret;

} 

int MemCache :: Decr(const char *data, unsigned dlen,  char **buf,unsigned &wsize, unsigned int &retv) 
{
 // hash key value
        unsigned uint_len = sizeof(uint32_t);
        if( dlen <= uint_len*2 )
        {
              DEBUG2("dlen <= uint_len*2 is true : dlen =", dlen);
              return PANDA_ERR_INPUT_DATA; 
        }
	unsigned ilen = 0;
        int ihash = 0;
	bcopy(data, &ihash, uint_len);
	ilen += uint_len;
        DEBUG4("ihash = ", ihash, "; offset = ", ilen); 
        
	int idx = ihash % m_iPartition;
	int iret = PANDA_MC_FAIL;
	_Lock(idx); 
	MemConn *p = (MemConn *)m_Conn_Map[idx];
	int istatus = p->pCon->getStatus();
	if(istatus == PANDA_MC_RESTART || istatus == PANDA_MC_LIVE )
	       istatus = iret = p->pCon->Decr(data + ilen, dlen - ilen, NULL , buf, wsize, retv);
	_UnLock(idx);
	if(PANDA_MC_NEXT == istatus)
	{
		idx = p->pCon->m_iBackup;
		_Lock(idx);
		long lt = p->pCon->getLastTime();
		p = (MemConn *)m_Conn_Map[idx];
		istatus = p->pCon->getStatus();   
		if( istatus == PANDA_MC_LIVE)
		{
			ostringstream os;
			os<< lt <<"|";
			istatus=iret=p->pCon->Decr(data+ilen, dlen-ilen, os.str().c_str(), buf, wsize, retv);
		}
		_UnLock(idx);
	}
	if(istatus == PANDA_MC_NEXT)
	{	DEBUG1("all is down, can't use !"); }
	return iret;
}

int MemCache :: Delete(const char *data, unsigned dlen ) //, char **buf, unsigned int &retv) 
{ //  hash + counts + key ....
        unsigned uint_len = sizeof(uint32_t);
        if( dlen <= uint_len*2 )
        {
              DEBUG2("dlen <= uint_len*2 is true : dlen =", dlen);
              return PANDA_ERR_INPUT_DATA; 
        }
	unsigned ilen = 0;
        int ihash = 0;
	bcopy(data, &ihash, uint_len);
	ilen += uint_len;
        DEBUG4("ihash = ", ihash, "; offset = ", ilen); 

        int argc = 0;
	bcopy(data + ilen , &argc, uint_len);
	ilen += uint_len;
        DEBUG4("argc = ", argc, "; offset = ", ilen); 
        if( argc <= 0)
        {
              DEBUG2("argc=0, means no key=value to save! ilen =", ilen); 
              return PANDA_ERR_INPUT_DATA; 
	}
	int idx = ihash % m_iPartition;
	int iret = PANDA_MC_FAIL;
	_Lock(idx); 
	MemConn *p = (MemConn *)m_Conn_Map[idx];
	int istatus = p->pCon->getStatus();
	if(istatus == PANDA_MC_RESTART || istatus == PANDA_MC_LIVE )
	{
		for(int i=0; i< argc; i++)
		{
			int klen;
			bcopy(data + ilen, &klen, uint_len);
			ilen += uint_len;
			if(ilen + klen > dlen || klen > PANDA_MC_KEY_MAX )
			{
				DEBUG4("ilen+klen>dlen; ilen=", ilen, "; klen=",klen); 
				return PANDA_ERR_INPUT_DATA;       
			}
			char *key = (char *)data + ilen;
			ilen += klen;
			istatus = iret = p->pCon->Delete( key, klen);
		}
	}
	_UnLock(idx);
	if(PANDA_MC_NEXT == istatus)
	{
		idx = p->pCon->m_iBackup;
		_Lock(idx);
		long lt = p->pCon->getLastTime();
		p = (MemConn *)m_Conn_Map[idx];
		istatus = p->pCon->getStatus();   
		if( istatus == PANDA_MC_LIVE)
		{

			for(int i=0; i< argc; i++)
			{
				int klen;
				bcopy(data + ilen, &klen, uint_len);
				ilen += uint_len;
				if(ilen + klen > dlen || klen>PANDA_MC_KEY_MAX)
				{
					DEBUG4("ilen+klen>dlen; ilen=", ilen, "; klen=",klen); 
					return PANDA_ERR_INPUT_DATA;       
				}
				char *key = (char *)data + ilen;
				ilen += klen;
				ostringstream os;
				char *nkey = g_strndup((char *)key, klen);
				os<< lt <<"|"<< nkey;
				g_free(nkey);
				istatus = iret = p->pCon->Delete(os.str().c_str(), os.str().length());
			}

		}
		_UnLock(idx);
	}
	if(istatus == PANDA_MC_NEXT)
	{	DEBUG1("all is down, can't use !");    }
        return iret;
}
	
        
int MemCache :: ResetCacheStatus(const char *ip, int port)
{
        int isize = m_Conn_Map.size();
        for(int idx = 0; idx < isize; idx++)
        {
                _Lock(idx);
                MemConn *p = (MemConn *)m_Conn_Map[idx];
                DEBUG4("check ", p->pCon->m_sHostIP.c_str(), ":", p->pCon->m_iPort);
                if(strcmp(p->pCon->m_sHostIP.c_str(), ip) == 0 && p->pCon->m_iPort == port)
                {
                        DEBUG5("Yes, set m_lLasttime = ", time(NULL),"(old =", p->pCon->m_lLasttime,"); m_bLive = true");
                        p->pCon->m_lLasttime = time(NULL);
                        p->pCon->m_bLive = true;
                }
                _UnLock(idx);
        }
        return 0;
}

void MemCache ::_Lock(int idx)
{
#ifdef _PTHREAD_SAFE
	MemConn *p = (MemConn *)m_Conn_Map[idx];
	pthread_mutex_lock(&p->lock);
#endif
}

void MemCache ::_UnLock(int idx)
{
#ifdef _PTHREAD_SAFE
	MemConn *p = (MemConn *)m_Conn_Map[idx];
	pthread_mutex_unlock(&p->lock);
#endif
}
